
#include <vector>
#include "../iface/debug.h"
#include "../utils/Hash.h"
#include "../utils/StringUtils.h"
#include "../net/Broker4.h"
#include "../crypt/RSAKeyPair.h"
#include "../net/SSHChain.h"
#include <apr-1/apr_strings.h>
#include "Broker4Server.h"

#define MAX_HEADERS 10
#define LIMIT_OF_ANY_MESSAGE 2097152

#define MAX_ENCRYPTED_HEADERS         20
#define MAX_ENCRYPTED_HEADER_LENGTH 8192

apr_table_t *Broker4Server::GetEncryptedHeaders(apr_table_t *plainheader, int i, apr_pool_t *mp, const char *buffer, apr_size_t length)
{
   if (length > MAX_ENCRYPTED_HEADER_LENGTH)
   {
      errprint("Encrypted headers are longer than permitted");
      return NULL;
   }
   const char *fromactor = NULL;
   const char *toactor = NULL;
   if ((fromactor = apr_table_get(plainheader, "FROM")) == NULL)
   {
      errprint("Could not find FROM header");
      return NULL;
   }
   if ((toactor = apr_table_get(plainheader, "TO")) == NULL)
   {
      errprint("Could not find TO header");
      return NULL;
   }
   KeyPair *pubkey_pair = dbs->keydb->GetPublicKeyByName(fromactor);
   KeyPair *prvkey_pair = dbs->keydb->GetPrivateKeyByName(toactor);
   if (prvkey_pair == NULL)
   {
      errprint("This program does not have access to private key '%s'\n", toactor);
      return NULL;
   }
   // Need to think about how to deenvelope signed data with a NULL key
   int wastrusted = 0;
   std::vector<char> *decrypt =
      Broker4::EncryptDataChunk(fromactor, toactor, pubkey_pair, prvkey_pair, buffer, length, 1, 1, mp, &wastrusted);
   // Need to think about what would happen if we removed all the limits on bucket brigades
   // - made them ubiquitous so that this returns a bucket brigade rather than an std::vector<char> (which is bad)
   // because a bucket brigade could do this next bit going into the header parser:
   // an std::vector<char> * can't!
   if (wastrusted == 0)
   {
      // Add special flag for untrusted sender,
      // change the FROM header in the original to NULL
      const char *unconfirmed = apr_table_get(plainheader, "FROM");
      apr_table_set(plainheader, "user", unconfirmed);
      apr_table_unset(plainheader, "FROM");
      apr_table_set(plainheader, "FROM", "NULL");
      // NULL is a special guest user with no priviliges
   }
}

int Broker4Server::DivineMeaning(BrokerMessage *msgin, BrokerMessage *msgout, int i, apr_pool_t *mp)
{
   char timebuf[APR_RFC822_DATE_LEN + 1];
   // returns 0 if meaning is obtained with a valid response
   // so should never return 1, but might do if error
   if (msgin->GetSender() == NULL) { errprint("No FROM field in message"); return 1; }
   if (msgin->GetRecipient() == NULL) { errprint("No TO field in message"); return 1; }
   const apr_table_t *htable;
   if ((htable = msgin->GetHeaderTable()) == NULL) { errprint("No headers field in message"); return 1; }
   msgout->ReplyTo(msgin);
   const char *processee;
   if ((msgin->IsType("REGISTER_USER_KEY")) && (msgin->IsPayloadCount(4)))
      if (((strcmp(msgin->GetProcessee(), &((msgin->GetSender())[1])) == 0) && ((msgin->GetSender())[0] == '.')) ||
          (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0))
      {
         Broker4::UserRegistration(dbs, StaticStringGetClientAddress(i), msgin, msgout);
         return 0;
      }
   if ((msgin->IsType("ADD_SSH_HOST")) && (msgin->IsPayloadCount(4)))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         int fingerprintfailed = 0;
         if (!(dbs->clustersdb->BooleanIsHostAccessibleByUser(msgin->GetProcessee(), msgin->GetPayload(1))))
            dbs->clustersdb->CreateLogInDetails(msgin->GetProcessee(), msgin->GetPayload(1), msgin->GetPayload(0));
         // May need to generate first key pair - and increment refcount to make permanent!
         std::vector<char *> *blacklist = new std::vector<char *>();
         std::vector<char *> *hostpath = dbs->clustersdb->GetPathFromSourceToHost(msgin->GetProcessee(), msgin->GetPayload(1), blacklist);
         SSHChain *tchain = NULL;
         while ((hostpath != NULL) && (tchain == NULL))
         {
            tchain = new SSHChain(dbs->clustersdb, mp);
            for (int i = 0; i < hostpath->size(); i++)
               tchain->AddToSSHChain(msgin->GetProcessee(), hostpath->at(i));
            tchain->SetAuthenticatorPassword(msgin->GetPayload(2));
            tchain->SetAuthenticatorKeys(dbs->sshkeydb, msgin->GetProcessee(), msgin->GetPayload(3));
            int didfail;
            SSHClient *scli = tchain->ActivateSSHChain(&didfail);
            if (didfail)
            {
               char *failedhost = tchain->GetFailedLinkHostName();
               errprint("failed to access '%s', blacklisting...", failedhost);
               blacklist->push_back(failedhost);
               tchain->DustForFingerprints(dbs->pendingdb);
               fingerprintfailed = tchain->WasFingerprintProblem();
               delete tchain;
               tchain = NULL;
               StringUtils::destroyStringList(hostpath);
               hostpath = dbs->clustersdb->GetPathFromSourceToHost(msgin->GetProcessee(), msgin->GetPayload(1), blacklist);
            }
         }
         char *allbrokenhosts = StringUtils::varprintf("");
         if ((blacklist) && (blacklist->size() > 0))
         {
            free(allbrokenhosts);
            allbrokenhosts = StringUtils::varprintf("%s", blacklist->at(0));
            for (int i = 1; i < blacklist->size(); i++)
            {
               char *oldbroke = allbrokenhosts;
               allbrokenhosts = StringUtils::varprintf("%s,%s", allbrokenhosts, blacklist->at(i));
               free(oldbroke);
            }
         }
         if (blacklist) StringUtils::destroyStringList(blacklist);
         if (hostpath == NULL)
         {
            msgout->AddTypeHeaderNOCRLF("SSH_ERROR", ((fingerprintfailed) ? "FINGERPRINT_UNTRUSTED" : "ROUTE_FAILED"));
            msgout->AddVerbatimPayloadString(allbrokenhosts);
            dbs->clustersdb->DeleteLogInDetails(msgin->GetProcessee(), msgin->GetPayload(1));
         }
         if (tchain)
         {
            if (hostpath) StringUtils::destroyStringList(hostpath);
            delete tchain;
            msgout->AddTypeHeaderNOCRLF("SSH_SUCCESS", msgin->GetProcessee());
            int keydb_needed_anchoring = dbs->keydb->EnsureChainBaseAnchored(msgin->GetProcessee());
            char *sshkeyowner = StringUtils::varprintf(".ssh_%s", msgin->GetProcessee());
            // Anchoring works on key name, not key owner
            int sshkeydb_needed_anchoring = dbs->sshkeydb->EnsureChainBaseAnchored(msgin->GetProcessee());
            if (keydb_needed_anchoring || sshkeydb_needed_anchoring)
            {
               char *latestsshkey = dbs->sshkeydb->GetLatestPublicKeyNameOfOwner(sshkeyowner);
               dbs->keydb->MarkKeyPairAsLatest(latestsshkey);
               free(latestsshkey);
               dbs->keydb->KeyPairNowUsedByActor(msgin->GetRecipient());
            }
            free(sshkeyowner);
         }
         if (allbrokenhosts) free(allbrokenhosts);
         return 0;
      }
   // No untrusted messages are allowed past this point!
   if ((!(msgin->GuaranteedAuthentic())) || ((msgin->GetSender())[0] == '.'))
   {
      msgout->AddTypeHeaderNOCRLF("ACCESS", "DENIED");
      return 0;
   }
   if ((msgin->IsType("REGISTER_COMPUTE")) && (msgin->IsPayloadCount(0)))
      if ((strcmp(msgin->GetProcessee(), msgin->GetSender()) == 0) &&
          (dbs->keydb->IsKeyIntendedForCompute(msgin->GetSender())))
      {
         dbs->keydb->KeyPairNowUsedByActor(msgin->GetRecipient());
         dbs->actorsdb->AddKeyDependency(msgin->GetSender(), msgin->GetRecipient());
         const char *durat = apr_table_get(htable, "Duration");
         dbs->actorsdb->AddComputeSession(msgin->GetSender(), ((durat) ? atoi(durat) : 300), pipearray[i].cliaddr, "single");
         msgout->AddTypeHeaderNOCRLF("COMPUTE_UUID", msgin->GetSender());
         return 0;
      }
   if ((msgin->IsType("BEGIN_SESSION")) && (msgin->GetHowManyPayloads() > 0))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         const char *restype = apr_table_get(htable, "Session-Type");
         int fail_delete = 0;
         char *jname = Hash::makeHexUUID_PlusUniq2(24, dbs->keydb, dbs->sshkeydb);
         switch (msgin->GetHowManyPayloads())
         {
            case 4:
               msgin->TakeAndInsertOwnedPublicKeyByName(dbs->keydb, 3, jname, "out_%s", jname);
            case 3:
               msgin->TakeAndInsertOwnedPrivateKeyByName(dbs->keydb, 2, jname, "out_%s", jname);
            case 2:
               msgin->TakeAndInsertOwnedPublicKeyByName(dbs->keydb, 1, jname, "in_%s", jname);
            default: break;
         }
         char *joutname = StringUtils::varprintf("out_%s", jname);
         dbs->keydb->KeyPairNowUsedByActor(joutname);
         char *jinname = StringUtils::varprintf("in_%s", jname);
         dbs->keydb->KeyPairNowUsedByActor(jinname);
         free(joutname);
         free(jinname);
         apr_off_t sizeused = 0;
         fail_delete |= Broker4::ProcessUploadedXMLDescriptor(dbs, jname, msgin, &sizeused);
         if (!fail_delete)
         {
            msgout->AddTypeHeaderNOCRLF("SESSION_ID", msgin->GetProcessee());
            msgout->AddHeaderLineNOCRLF("UsedSpace: %" APR_OFF_T_FMT, sizeused);
            msgout->AddVerbatimPayloadString(jname);
            msgout->AttachPublicKeyByName(dbs->keydb, jname);
         } else
         {
            // this roll back crashed, I want to know why at some point...
            DBArbitrator::DeleteFileStoreByUUID(dbs, jname, mp);
         }
         free(jname);
         return 0;
      }
   if ((msgin->IsType("EXPUNGE_SESSION")) && (msgin->IsPayloadCount(0)))
      if ((dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0) &&
          (dbs->actorsdb->VerifyUUIDIsWorkStore(msgin->GetProcessee()) == 0))
      {
         KeyPair *saved_kp = dbs->keydb->GetPrivateKeyByName(msgin->GetProcessee());
         // Must find subsidiary filestores also...
         std::vector<char *> *subsidiaries = dbs->actorsdb->ViewWorkStoresOwned(msgin->GetProcessee());
         if (subsidiaries)
         {
            for (int i = 0; i < subsidiaries->size(); i++)
               DBArbitrator::DeleteFileStoreByUUID(dbs, subsidiaries->at(i), mp);
            StringUtils::destroyStringList(subsidiaries);
         }
         DBArbitrator::DeleteFileStoreByUUID(dbs, msgin->GetProcessee(), mp);
         msgout->ForceDestroySendingKey();
         dbs->keydb->AddPrivateKeyWithPassword(msgin->GetProcessee(), dbs->keydb->WIPpassword, saved_kp, NULL);
         delete saved_kp;
         msgout->AddTypeHeaderNOCRLF("SESSION_EXPUNGED", msgin->GetProcessee());
         return 0;
      }
   if ((msgin->IsType("UPLOAD_OKAY")) && (msgin->IsPayloadCount(0)))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         const char *bnumber = apr_table_get(htable, "Block-Number");
         if (bnumber) { Broker4::CheckFileComplete(msgin->GetProcessee(), atoi(bnumber), msgout); return 0; }
      }
   if ((msgin->IsType("UPLOAD_DO")) && (msgin->IsPayloadCount(2)))
      if ((dbs->actorsdb->QuietlyVerifyUUIDIsCompute(msgin->GetSender()) == 0) ||
          (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0))
      {
         const char *bnumber = apr_table_get(htable, "Block-Number");
         if (bnumber) { Broker4::WriteFileAndCheckHash(dbs, atoi(bnumber), msgin, msgout); return 0; }
      }
   if (msgin->IsType("HAPPY_COMPUTE"))
      if (dbs->actorsdb->VerifyUUIDIsCompute(msgin->GetSender()) == 0)
      {
         errprint("got happy message");
         apr_status_t status;
         apr_time_t timet = apr_time_now();
         status = apr_rfc822_date(timebuf, timet);
         CheckAPRError(status);
         const apr_table_t *htable = msgin->GetHeaderTable();
         const char *wdatatag = apr_table_get(htable, "WorkDataTag");
         if (wdatatag != NULL)
            dbs->actorsdb->sqlInstantExecf("update 'workdata' set lasthappydate='%s' where name='%s';", timebuf, wdatatag);
         std::vector<char *> *wk_info = dbs->actorsdb->sqlScanRowTextf("select jobclusterid from 'workdata' where name='%s';", wdatatag);
         if ((wk_info == NULL) || (wk_info->size() == 0))
         {
            errprint("happy compute message received from workdata that does not exist or has no associated cluster!");
            if (wk_info) delete wk_info;
         } else
         {
            dbs->clustersdb->sqlInstantExecf("update 'clusters' set lasthappydate='%s' where name='%s';", timebuf, wk_info->at(0));
            StringUtils::destroyStringList(wk_info);
         }
         if (dbs->actorsdb->IsWorkStoreStarted(msgin->GetProcessee()) == -1)
            msgout->AddTypeHeaderNOCRLF("WORK_CANCELLED", msgin->GetProcessee());
         else
            msgout->AddTypeHeaderNOCRLF("HAPPY_BROKER", msgin->GetProcessee());
         return 0;
      }
   if ((msgin->IsType("SET_ACTIVE")) && (msgin->IsPayloadCount(1)))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         int int_instances = 1;
         int int_duration = 20;
         int positive_response = 0;
         int workstorestatus = dbs->actorsdb->IsWorkStoreStarted(msgin->GetProcessee());
         if (workstorestatus > 0)
         {
            const char *str_instances = NULL;
            if ((str_instances = apr_table_get(htable, "Instances")) != NULL) int_instances = atoi(str_instances);
            const char *str_duration = NULL;
            if ((str_duration = apr_table_get(htable, "Max-Time")) != NULL) int_duration = atoi(str_duration);
            positive_response = Broker4::TryToStartWorkStore(dbs, msgin->GetProcessee(), int_instances, int_duration, msgin->GetPayload(0), apr_table_get(htable, "ClusterIDs"), mp);
         } else if (workstorestatus == -1)
            errprint("Work store '%s' does not exist/unknown error", msgin->GetProcessee());
         if (positive_response == 0)
         {
            msgout->AddTypeHeaderNOCRLF("WORK_INACTIVE", msgin->GetProcessee());
            return 0;
         }
         if ((positive_response == 1) || (workstorestatus == 1) || (workstorestatus == 0))
         {
            msgout->AddTypeHeaderNOCRLF("WORK_ACTIVATED", msgin->GetProcessee());
            // this may not reflect the true number of instances if the work is started differently
            msgout->AddHeaderLineNOCRLF("Copies: %d", int_instances);
            return 0;
         }
      }
   if ((msgin->IsType("SET_USABLE")) && (msgin->IsPayloadCount(0)))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         int positive_reponse = 0;
         int workstorestatus = dbs->actorsdb->IsWorkStoreStarted(msgin->GetProcessee());
         if (workstorestatus > 0) dbs->actorsdb->WorkStoreSetUsable(msgin->GetProcessee());
         if (workstorestatus == -1) errprint("Work store '%s' does not exist/unknown error", msgin->GetProcessee());
         if (workstorestatus != -1)
         {
            msgout->AddTypeHeaderNOCRLF("WORK_USABLE", msgin->GetProcessee());
            return 0;
         }
      }
   if ((msgin->IsType("REQUEST_WORK")) && (msgin->IsPayloadCount(0)))
      if (dbs->actorsdb->VerifyUUIDIsCompute(msgin->GetSender()) == 0)
      {
         char *ultuser = DBArbitrator::LookupComputeUltimateUser(dbs, msgin->GetSender());
         const char *given_tmax_str = apr_table_get(htable, "Time-Remaining");
         std::vector<char *> *twork = dbs->actorsdb->AtomicGetWork(ultuser, given_tmax_str ? atoi(given_tmax_str) : INT_MAX);
         free(ultuser);
         if ((twork) && (twork->size() > 0))
         {
            msgout->AddTypeHeaderNOCRLF("WORK_ASSIGNMENT", twork->at(3));
            msgout->AddHeaderLineNOCRLF("Max-Time: %s", twork->at(1));
            msgout->AddHeaderLineNOCRLF("Local-Ident: %s", twork->at(2));
            msgout->AddHeaderLineNOCRLF("Total-Instances: %d", dbs->actorsdb->GetInstancesFromWorkStore(twork->at(3)));
            msgout->AddHeaderLineNOCRLF("Block-Count: %d", dbs->actorsdb->NumberOfWorkStoreBlocksFromUUID(twork->at(3)));
            msgout->AddHeaderLineNOCRLF("WorkDataTag: %s", twork->at(0));
            msgout->AttachPublicKeyByName(dbs->keydb, twork->at(3));
            msgout->AttachPrivateKeyByName(dbs->keydb, twork->at(3));
            msgout->AttachPrivateKeyByName(dbs->keydb, "out_%s", twork->at(3));
            msgout->AttachPublicKeyByName(dbs->keydb, "in_%s", twork->at(3));
            msgout->AttachEncryptedXMLDescriptorByName(dbs->keydb, twork->at(3), "_%s.xml", twork->at(3));
            StringUtils::destroyStringList(twork);
         } else
         {
            msgout->AddTypeHeaderNOCRLF("WORKUNAVAILABLE", msgin->GetSender());
            if (twork) delete twork;
         }
         return 0;
      }
   if ((msgin->IsType("DOWNLOAD_DO")) && (msgin->IsPayloadCount(0)))
      if ((strcmp(&((msgin->GetSender())[4]), msgin->GetProcessee()) == 0) ||
          (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0))
      {
         const char *bnumber = apr_table_get(htable, "Block-Number");
         if (bnumber)
         {
            msgout->AddTypeHeaderNOCRLF("DOWNLOAD_BLOCK", msgin->GetProcessee());
            const apr_table_t *htable = msgin->GetHeaderTable();
            const char *wdatatag = apr_table_get(htable, "WorkDataTag");
            if (wdatatag != NULL)
               dbs->actorsdb->sqlInstantExecf("update 'workdata' set blocksuploaded='%d' where name='%s';", atoi(bnumber)+1, wdatatag);
            apr_size_t fullsize = msgout->AttachHashThenFileByName("_%s.f%d", msgin->GetProcessee(), atoi(bnumber));
            return 0;
         }
      }
   if ((msgin->IsType("QUERY_STATE")) && (msgin->IsPayloadCount(0)))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         std::vector<char *> *hlist = dbs->actorsdb->GetUserState(msgin->GetProcessee(), mp);
         msgout->AddTypeHeaderNOCRLF("ALLWORK", msgin->GetSender());
         msgout->AddHeaderLineNOCRLF("QuotaBytes: %" APR_OFF_T_FMT, dbs->actorsdb->GetUserQuota(msgin->GetProcessee()));
         if (hlist)
         {
            const char *flatstr = StringUtils::flattenStringList(hlist, ",");
            msgout->AddVerbatimPayloadString(flatstr);
            free((void *)flatstr);
            StringUtils::destroyStringList(hlist);
         }
         return 0;
      }
   if ((msgin->IsType("QUERY_WORKDATA_STATE")) && (msgin->IsPayloadCount(0)))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         char *ultuser = dbs->actorsdb->LookupUltimateUser(msgin->GetSender());
         std::vector<char *> *hlist = dbs->actorsdb->GetUserWorkDataState(ultuser, msgin->GetProcessee(), mp);
         msgout->AddTypeHeaderNOCRLF("ALLWORKDATA", msgin->GetSender());
         if (hlist)
         {
            const char *flatstr = StringUtils::flattenStringList(hlist, ",");
            msgout->AddVerbatimPayloadString(flatstr);
            free((void *)flatstr);
            StringUtils::destroyStringList(hlist);
         }
         return 0;
      }
   if ((msgin->IsType("RESULTS_READY")) && (msgin->IsPayloadCount(0)))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         std::vector<char *> *hlist = dbs->actorsdb->GetReadyResults(msgin->GetProcessee());
         if (hlist)
         {
            const char *flatstr = StringUtils::flattenStringList(hlist, ",");
            msgout->AddTypeHeaderNOCRLF("RESULTS_DOWNLOADABLE", msgin->GetSender());
            msgout->AddVerbatimPayloadString(flatstr);
            free((void *)flatstr);
            StringUtils::destroyStringList(hlist);
            return 0;
         } else
         {
            msgout->AddTypeHeaderNOCRLF("NO_RESULTS", msgin->GetSender());
            return 0;
         } 
      }
   if ((msgin->IsType("REQUEST_UPDATE_KEYS")) && (msgin->IsPayloadCount(0)))
   {
      char *latest_key_owned = dbs->keydb->GetLatestPublicKeyNameOfOwner(dbs->keydb->selfkey);
      if (latest_key_owned)
      {
         if (strcmp(msgin->GetRecipient(), latest_key_owned) == 0)
         {
            msgout->AddTypeHeaderNOCRLF("NO_UPDATE", msgin->GetProcessee());
         } else
         {
            msgout->AddTypeHeaderNOCRLF("KEY_UPDATE", dbs->keydb->selfkey);
            msgout->AddVerbatimPayloadString(latest_key_owned);
            msgout->AttachPublicKeyByName(dbs->keydb, latest_key_owned);
            // This is referring to the later BROKER key
            dbs->keydb->KeyNowUnusedByActor(msgin->GetRecipient());
            dbs->keydb->KeyNowDeletedByActor(msgin->GetRecipient());
         }
      } else errprint("could not find latest key with owner '%s'", dbs->keydb->selfkey);
      free(latest_key_owned);
      return 0;
   }
   if ((msgin->IsType("KEY_UPDATE")) && (msgin->IsPayloadCount(2)))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         char *latest_key_owned = dbs->keydb->GetLatestPublicKeyNameOfOwner(msgin->GetProcessee());
         if ((latest_key_owned) && (strcmp(msgin->GetSender(), latest_key_owned) == 0))
         {
            if (dbs->keydb->GetPublicKeyUsedCount(msgin->GetProcessee()) == 0)
            {
               msgout->AddTypeHeaderNOCRLF("UPDATE_REFUSED", msgin->GetProcessee());
               return 0;
            }
            StringUtils::aggressiveStringSanitiser((char *)msgin->GetPayload(0));
            // Key update seems genuine - perform update!
            msgin->TakeAndInsertOwnedPublicKeyByName(dbs->keydb, 1, msgin->GetProcessee(), msgin->GetPayload(0));
            // No keys but broker keys require refcounting before deletion!
            // We do not actually want to delete the user key -
            // if we do, nothing prevents someone else registering the same name and having the same job permissions
            
            // I'm not sure this works at all
            /*
            if (strcmp(msgin->GetProcessee(), latest_key_owned) != 0)
            {
               dbs->keydb->KeyNowUnusedByActor(latest_key_owned);
               dbs->keydb->KeyNowDeletedByActor(latest_key_owned);
            }
            */
            // Possible scenario where you are the only user, you update your broker key to one that has been around
            // for x time with no-one on it, broker key updates,
            // and GC kills the old key before you can place your token on it. I think this will never happen though.
            // SOLUTION - make the 0 token GC run at opposite times to the key updates
            dbs->keydb->KeyPairNowUsedByActor(msgin->GetRecipient());
            msgout->SetRecipient(msgin->GetPayload(0));
            dbs->actorsdb->AddOrReplaceDelegateKey(msgin->GetPayload(0), msgin->GetProcessee());
            msgout->AddTypeHeaderNOCRLF("KEYS_UPDATE_OKAY", msgin->GetProcessee());
         }
         if (latest_key_owned) free(latest_key_owned);
         return 0;
      }
   if ((msgin->IsType("GET_CLUSTERS")) && (msgin->IsPayloadCount(0)))
      if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      {
         char *access_clusters = dbs->clustersdb->GetListOfClustersForUser(msgin->GetProcessee());
         if ((access_clusters) && (strlen(access_clusters) > 0))
         {
            msgout->AddTypeHeaderNOCRLF("ACCESSIBLE_CLUSTERS", msgin->GetProcessee());
            msgout->AddVerbatimPayloadString(access_clusters);
            free(access_clusters);
         } else
         {
            if (access_clusters) free(access_clusters);
            msgout->AddTypeHeaderNOCRLF("NO_CLUSTERS", msgin->GetProcessee());
         }
         return 0;
      }
   printf("UNRECOGNISED MESSAGE:\n");
   msgin->Print();
   if (dbs->actorsdb->GrantUUIDPermission(msgin->GetSender(), msgin->GetProcessee()) == 0)
      printf("UUID permissions okay\n");
   else
      printf("(%s,%s)\n", msgin->GetSender(), msgin->GetProcessee());
   printf("%d\n", msgin->GetHowManyPayloads());
   msgin->Print();
   msgout->AddTypeHeaderNOCRLF("ACCESS", "DENIED");
   return 0;
}

void Broker4Server::GenerateConsole(int i, apr_pool_t *mp)
{
   static const char *html_start = "<html><head><title>Console</title></head>" CRLF_STR;
   static const char *open_form = "<body><form name=\"permissions\" action=\"console.html\" method=\"POST\">" CRLF_STR;
   static const char *open_div = "<div align=\"center\"><br />" CRLF_STR;
   static const char *individ_header = "<h2>%s</h2>" CRLF_STR;
   static const char *user_record = "'%s' at %s from %s with %s &nbsp; " CRLF_STR;
   static const char *keyup_record = "'%s' expected host fingerprint '%s', obtained '%s' &nbsp; " CRLF_STR;
   static const char *input_radio = "<input type=\"radio\" name=\"%s\" value=\"%s\" />%s" CRLF_STR;
   static const char *input_cradio = "<input type=\"radio\" name=\"%s\" value=\"%s\" checked />%s" CRLF_STR;
   static const char *next_line = "<br />" CRLF_STR;
   static const char *submit_button = "<br /><br /><input type=\"submit\" action=\"console.html\" value=\"Submit Changes\" />" CRLF_STR;
   static const char *html_stop = "</div></form></body></html>" CRLF_STR CRLF_STR;
   static const char *firstheader = "HTTP/1.0 200 OK" CRLF_STR CRLF_STR;
   pipearray[i].handler_to_send->writeToStream((char *)firstheader, strlen(firstheader));
   pipearray[i].handler_to_send->writeToStream((char *)html_start, strlen(html_start));
   pipearray[i].handler_to_send->writeToStream((char *)open_form, strlen(open_form));
   pipearray[i].handler_to_send->writeToStream((char *)open_div, strlen(open_div));
   std::vector<char *> *pendingusers = dbs->pendingdb->GetAllPendingUserRequests();
   // 4 field to the pending users database
#define COLUMNS_USERS_FIELDS 5
   int countuserrows = pendingusers->size() / COLUMNS_USERS_FIELDS;
   if (countuserrows > 0)
   {
      pipearray[i].handler_to_send->writefToStream((char *)individ_header, "Unauthorised Users");
      for (int k = 0; k < countuserrows; k++)
      {
         RSAKeyPair rsa;
         rsa.ImportPublicKey(pendingusers->at(k*COLUMNS_USERS_FIELDS+3));
         char *long_fingerprint = rsa.GetMD5KeyFingerprint();
         char *short_fingerprint = rsa.GetFingerprintAsVariableName();
         if (long_fingerprint && short_fingerprint)
         {
            pipearray[i].handler_to_send->writefToStream((char *)user_record,
               pendingusers->at(k*COLUMNS_USERS_FIELDS+0),
               pendingusers->at(k*COLUMNS_USERS_FIELDS+2),
               pendingusers->at(k*COLUMNS_USERS_FIELDS+1),
               long_fingerprint);
            pipearray[i].handler_to_send->writefToStream((char *)input_radio, short_fingerprint, "Allow", "Allow");
            pipearray[i].handler_to_send->writefToStream((char *)input_radio, short_fingerprint, "Deny", "Deny");
            pipearray[i].handler_to_send->writefToStream((char *)input_cradio, short_fingerprint, "Ignore", "Ignore");
            pipearray[i].handler_to_send->writeToStream((char *)next_line, strlen(next_line));
         }
         if (short_fingerprint) free(short_fingerprint);
         if (long_fingerprint) free(long_fingerprint);
      }
   }
   std::vector<char *> *pendingkeyupdates = dbs->pendingdb->GetAllPendingHostKeyUpdates();
#define COLUMNS_KEYUPDATE_FIELDS 3
   int countkeyupdaterows = pendingkeyupdates->size() / COLUMNS_KEYUPDATE_FIELDS;
   if (countkeyupdaterows > 0)
   {
      pipearray[i].handler_to_send->writefToStream((char *)individ_header, "Host fingerprints");
      for (int k = 0; k < countkeyupdaterows; k++)
      {
         const char *t_hostname = pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+0);
         const char *t_oldfp = pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+1);
         const char *t_newfp = pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+2);
         int needs_none = ((t_oldfp == NULL) || (strlen(t_oldfp) == 0));
         // I can ,make this nicer to look at later! - format the key fingerprints the same!
         pipearray[i].handler_to_send->writefToStream((char *)keyup_record, t_hostname, needs_none ? t_oldfp : "(none)", t_newfp);
         char *key_fingerprint_as_variable_name = (char *)malloc(sizeof(char)*(strlen(t_newfp)+1+1));
         key_fingerprint_as_variable_name[0] = '_';
         strcpy(&(key_fingerprint_as_variable_name[1]), t_newfp);
         pipearray[i].handler_to_send->writefToStream((char *)input_radio, key_fingerprint_as_variable_name, "Allow", "Allow");
         pipearray[i].handler_to_send->writefToStream((char *)input_radio, key_fingerprint_as_variable_name, "Deny", "Deny");
         pipearray[i].handler_to_send->writefToStream((char *)input_cradio, key_fingerprint_as_variable_name, "Ignore", "Ignore");
         pipearray[i].handler_to_send->writeToStream((char *)next_line, strlen(next_line));
         free(key_fingerprint_as_variable_name);
      }
   }
   if ((countuserrows > 0) || (countkeyupdaterows > 0))
      pipearray[i].handler_to_send->writeToStream((char *)submit_button, strlen(submit_button));
   pipearray[i].handler_to_send->writeToStream((char *)html_stop, strlen(html_stop));
   pipearray[i].handler_to_send->writeEOS();
   StringUtils::destroyStringList(pendingusers);
   StringUtils::destroyStringList(pendingkeyupdates);
}

void Broker4Server::UnderstandPOSTData(apr_table_t *data)
{
   std::vector<char *> *pendingusers = dbs->pendingdb->GetAllPendingUserRequests();
   // 4 field to the pending users database
#define COLUMNS_USERS_FIELDS 5
   int countuserrows = pendingusers->size() / COLUMNS_USERS_FIELDS;
   if (countuserrows > 0)
   {
      for (int k = 0; k < countuserrows; k++)
      {
         RSAKeyPair rsa;
         rsa.ImportPublicKey(pendingusers->at(k*COLUMNS_USERS_FIELDS+3));
         char *short_fingerprint = rsa.GetFingerprintAsVariableName();
         if (short_fingerprint)
         {
            const char *tableval = apr_table_get(data, short_fingerprint);
            if (tableval)
            {
               char *temp_user = (char *)StringUtils::varprintf(".%s", pendingusers->at(k*COLUMNS_USERS_FIELDS+0));
//               if (strcmp(tableval, "Allow") == 0) Broker4::AcceptUserRegistration(dbs, temp_user);
//               if (strcmp(tableval, "Deny") == 0) Broker4::DenyUserRegistration(dbs, temp_user);
//               if ((strcmp(tableval, "Allow") == 0) || (strcmp(tableval, "Deny") == 0)) Broker4::CleanUpUserRegistration(dbs, temp_user);
               free(temp_user);
            }
         }
         if (short_fingerprint) free(short_fingerprint);
      }
   }
   StringUtils::destroyStringList(pendingusers);
   std::vector<char *> *pendingkeyupdates = dbs->pendingdb->GetAllPendingHostKeyUpdates();
#define COLUMNS_KEYUPDATE_FIELDS 3
   int countkeyupdaterows = pendingkeyupdates->size() / COLUMNS_KEYUPDATE_FIELDS;
   if (countkeyupdaterows > 0)
   {
      for (int k = 0; k < countkeyupdaterows; k++)
      {
         const char *t_newfp = pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+2);
         char *key_fingerprint_as_variable_name = (char *)malloc(sizeof(char)*(strlen(t_newfp)+1+1));
         key_fingerprint_as_variable_name[0] = '_';
         strcpy(&(key_fingerprint_as_variable_name[1]), t_newfp);
         const char *tableval = apr_table_get(data, key_fingerprint_as_variable_name);
         if (tableval)
         {
            if (strcmp(tableval, "Allow") == 0)
               dbs->clustersdb->UpdateKeyFingerprint(t_newfp, pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+0));
            if ((strcmp(tableval, "Allow") == 0) || (strcmp(tableval, "Deny") == 0))
               dbs->pendingdb->RemovePendingHostKeyUpdate(pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+0));
         }
         free(key_fingerprint_as_variable_name);
      }
   }
   StringUtils::destroyStringList(pendingkeyupdates);
}

int Broker4Server::ServeStreamFromSystemList(apr_table_t *htable, int i, apr_pool_t *mp, const char **systemwhitelist)
{
   char *dstr = apr_pstrdup(mp, apr_table_get(htable, "GET"));
   char *httpver = NULL;
   char *firsttok = apr_strtok(dstr, " ", &httpver);
   int count = 0;
   int passes = 0;
   // see if it matches any in the white list
   while (systemwhitelist[count] != NULL)
      passes |= (strcmp(systemwhitelist[count++], firsttok) == 0);
   if (passes == 0) return 0;
   if (strcmp(StaticStringGetClientAddress(i), "127.0.0.1") != 0)
      return 0;
   GenerateConsole(i, mp);
   return 1;
}

void Broker4Server::GETSession(int i, apr_pool_t *mp)
{
   static const char *systemwhitelist[] = { "/console.html", NULL };
   pipearray[i].waiting->lock();
   apr_atomic_inc32(&(pipearray[i].stage));
   apr_table_t *headertable = GetHeaders(i, mp);
   apr_atomic_inc32(&(pipearray[i].stage));
   if (headertable == NULL) { Error404(i); errprint("Header table missing (out of memory?)."); return; }
   if (apr_is_empty_table(headertable)) { Error404(i); errprint("Header table empty (unknown error)."); return; }
   std::vector<char *> *vec_whitelist = new std::vector<char *>();
   vec_whitelist->push_back(StringUtils::varprintf("/hosted_files/brokerkeyname"));
   vec_whitelist->push_back(StringUtils::varprintf("/hosted_files/bootstrap.sh"));
   vec_whitelist->push_back(StringUtils::varprintf("/hosted_files/queue.sh"));
   vec_whitelist->push_back(StringUtils::varprintf("/hosted_files/broker.pub"));
   vec_whitelist->push_back(StringUtils::varprintf("/hosted_files/compute"));
   char *latestkey = dbs->keydb->GetLatestPublicKeyNameOfOwner(BROKERNAME);
   if (latestkey)
   {
      vec_whitelist->push_back(StringUtils::varprintf("/hosted_files/%s.pub", latestkey));
      free(latestkey);
   } else errprint("could not find latest key owned by '%s'", BROKERNAME);
   vec_whitelist->push_back(NULL);
   if (ServeFilesFromWhiteList(headertable, i, mp, (const char **)(&(vec_whitelist->at(0)))))
   {
      IncrementUnlock(i);
      StringUtils::destroyStringList(vec_whitelist);
      return;
   }
   StringUtils::destroyStringList(vec_whitelist);
   if (ServeStreamFromSystemList(headertable, i, mp, systemwhitelist)) { IncrementUnlock(i); return; }
   errprint("Can not match request for %s", apr_table_get(headertable, "GET"));
   Error404(i);
}

void Broker4Server::POSTSession(int i, apr_pool_t *mp)
{
   if (strcmp(StaticStringGetClientAddress(i), "127.0.0.1") != 0)
   {
      errprint("erroneous post by %s", StaticStringGetClientAddress(i));
      pipearray[i].handler_to_send->writeEOS();
      apr_atomic_set32(&(pipearray[i].stage), SERVERSTAGE_SENDING);
      return;
   }
   pipearray[i].waiting->lock();
   apr_atomic_inc32(&(pipearray[i].stage));
   apr_table_t *headertable = GetHeaders(i, mp);
   apr_size_t contentlength = GetContentLength(headertable);
   apr_table_t *posttable = GetPOSTData(i, mp, contentlength);
   apr_atomic_inc32(&(pipearray[i].stage));
   UnderstandPOSTData(posttable);
   RedirectAfterPOST(i, mp, "/console.html");
   apr_atomic_inc32(&(pipearray[i].stage));
   pipearray[i].waiting->unlock();
}

void Broker4Server::BrokerSession(int i, apr_pool_t *mp)
{
   pipearray[i].waiting->lock();
   apr_atomic_inc32(&(pipearray[i].stage));
   BrokerMessage msg(mp);
   if (!Broker4::ReadEncMessageFromBB(pipearray[i].recv_to_handler, &msg, dbs->keydb, /*mp, */pipearray[i].waiting))
   {
      apr_atomic_inc32(&(pipearray[i].stage));
      BrokerMessage sendmsg(mp);
      //Broker4::NullAll(&sendmsg);
      if (!DivineMeaning(&msg, &sendmsg, i, mp))
      {
         if (!Broker4::WriteEncMessageToBB(pipearray[i].handler_to_send, &sendmsg, dbs->keydb/*, mp*/))
         {
            fprintf(stdout, ".");
            fflush(stdout);
            if ((sendmsg.GetRecipient())[0] == '.') dbs->keydb->DeletePublicKeyByName(sendmsg.GetRecipient());
            if (sendmsg.WeMustDestroySendingKey()) dbs->keydb->DeletePrivateKeyByName(sendmsg.GetSender());
         } else errprint("Could not write reply");
      } else errprint("Did not understand the message!");
//      Broker4::DeleteAll(&sendmsg);
   } else
   {
      apr_atomic_inc32(&(pipearray[i].stage));
      errprint("Could not read the message");
   }
   pipearray[i].handler_to_send->writeEOS();
//   Broker4::DeleteAll(&msg);
   apr_atomic_inc32(&(pipearray[i].stage));
   pipearray[i].waiting->unlock();
   // perhaps we need to get content-length and then call DivineMeaning
   // earlier - after having received >= content-length of body
}

void *Broker4Server::ClientHandler(apr_thread_t *tthread, void *args)
{
   apr_status_t status;
   Broker4Server *kthis = *((Broker4Server **)(((void **)(args))[0]));
   int i = *((int *)(((void **)(args))[1]));
   apr_pool_t *upperpool, *mp;
   upperpool = apr_thread_pool_get(tthread);
   status = apr_pool_create(&mp, upperpool);
   CheckAPRError(status);

   while (!(kthis->sessionstop))
   {
      kthis->pipearray[i].sessionstart->wait();
      if (kthis->sessionstop) break;
      apr_pool_t *sp;
      apr_pool_create(&sp, mp);
      while ((kthis->pipearray[i].recv_to_handler->realLength() < 4) &&
             (!(kthis->pipearray[i].recv_to_handler->isClosed())))
         apr_thread_yield();
      char fst[5];
      kthis->pipearray[i].recv_to_handler->readFromStream(fst, 4); fst[4] = '\0';
      if (!(kthis->pipearray[i].recv_to_handler->isClosed()))
      {
         if (strcmp(fst, "GET ") == 0)
         {
            kthis->GETSession(i, mp);
         } else if (strcmp(fst, "POST") == 0)
         {
            kthis->POSTSession(i, mp);
         } else
         {
            kthis->BrokerSession(i, mp);
         }
      } else
      {
         apr_atomic_set32(&(kthis->pipearray[i].stage), SERVERSTAGE_SENDING);
         kthis->pipearray[i].handler_to_send->writeEOS();
      }
      apr_pool_clear(sp);
      apr_pool_destroy(sp);
   }
   kthis->pipearray[i].sessionstart->unlock();
   apr_pool_clear(mp);
   apr_pool_destroy(mp);
   ERR_remove_state(0);
   apr_thread_exit(tthread, APR_SUCCESS);
   return NULL;
}
